package com.boboPlanet.comm.rabbitmq.factory;

import com.boboPlanet.comm.context.BusinessThreadContextHolder;
import com.boboPlanet.comm.context.LoginUserContext;
import com.boboPlanet.comm.entity.dto.PlanetMqMsg;
import com.boboPlanet.comm.enums.PlanetCommCode;
import com.boboPlanet.comm.exception.BusinessException;
import com.boboPlanet.comm.rabbitmq.producer.IMqBuild;
import com.boboPlanet.comm.thread.ThreadMdcUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;

/**
 * @Description: TODO topic factory msg
 * @Author: mis_wu
 * @Date: 2023/6/27
 * @Company planet
 **/
@Slf4j
public class TopicExchangeHandler {

    private final String queue;

    private final String exchange;

    private final String routingKey;

    private String msgId;

    private final Object data;

    private Boolean isAuth;

    private final RabbitTemplate rabbitTemplate;

    public TopicExchangeHandler(RabbitTemplate rabbitTemplate,IMqBuild mqBuild,Object data){
        this.rabbitTemplate = rabbitTemplate;
        this.queue = mqBuild.queue();
        this.exchange = mqBuild.exchange();
        this.routingKey = mqBuild.routingKey();
        this.isAuth = mqBuild.isAuth();
        this.data = data;

        topicExchangeSendMsg();
    }

    public void topicExchangeSendMsg(){
        msgId = UUID.randomUUID().toString().replaceAll("-","");

        this.assertMessage();

        log.info("rabbit mq topic queue is: {} ,exchange is: {} ,routingKey is: {} ,msg data is: {}, msgId is: {}",queue,exchange,routingKey,data,msgId);

        Message message = messageHandle();

        CorrelationData correlationData = new CorrelationData(msgId);
        correlationData.getFuture().addCallback(
                result -> {
                    assert result != null;
                    if (result.isAck()) {
                        log.info("消息发送成功, msgId:{}",correlationData.getId());
                    }else {
                        log.error("消息发送失败, msgId: {}, 失败原因: {}",correlationData.getId(),result.getReason());
                    }
                },
                ex -> {
                    log.error("消息发送异常, msgId: {}, 失败原因: {}",correlationData.getId(),ex.getMessage());
                }
        );

        rabbitTemplate.convertAndSend(exchange,routingKey,message,correlationData);

    }

    private void assertMessage(){
        if (data == null){
            throw BusinessException.build("MQ.data 消息内容不能为空");
        }
        if (StringUtils.isBlank(queue)){
            throw BusinessException.build("MQ.queue 消息队列不能为空");
        }
        if (StringUtils.isBlank(exchange)){
            throw BusinessException.build("MQ.exchange 消息交换机不能为空");
        }
        if (StringUtils.isBlank(routingKey)){
            throw BusinessException.build("MQ.routingKey 消息路由key不能为空");
        }
        if (StringUtils.isBlank(msgId)){
            throw BusinessException.build("MQ.msgId 消息ID不能为空");
        }
    }

    private Message messageHandle(){
        //MQ异步，需要注入线程上下文信息
        BusinessThreadContextHolder capture = BusinessThreadContextHolder.capture();

        PlanetMqMsg mqMsg = new PlanetMqMsg();
        mqMsg.setQueue(queue);
        mqMsg.setExchange(exchange);
        mqMsg.setRoutingKey(routingKey);
        mqMsg.setMessageId(msgId);
        mqMsg.setLoginUserInfo(LoginUserContext.getLoginUserInfoPO());
        //灰度标识待定
        mqMsg.setGray("");
        mqMsg.setData(data);

        MessageProperties properties = new MessageProperties();

        //注入线程上下文
        ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
        try {
            ObjectOutputStream out = new ObjectOutputStream(byteOutStream);
            out.writeObject(capture);
            out.flush();
        } catch (IOException e) {
            log.error("topic mq serializable fail,reason: {}",e.getMessage(),e);
            throw new RuntimeException(e);
        }
        List<Boolean> auth = Arrays.asList(true, false);
        if (isAuth == null || !auth.contains(isAuth)){
            isAuth = true;
        }
        //MQ授权设置
        properties.setHeader(PlanetCommCode.MQ_HEADER_AUTHORIZATION.getCode(),isAuth);

        properties.setHeader(PlanetCommCode.COMM_LOG_KEY.getCode(), ThreadMdcUtil.get());

        properties.setHeader(PlanetCommCode.MQ_HEADER_CAPTURE.getCode(),byteOutStream.toByteArray());

        return MessageBuilder.withBody(mqMsg.serialize().getBytes())
                //消息持久化
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .copyProperties(properties).build();

        //return new Message(mqMsg.serialize().getBytes(),properties);
    }
}
